package com.gengzp.mq.server.manager.impl;

import com.gengzp.mq.server.manager.MqBlockWaitingManager;
import com.gengzp.mq.server.model.message.entity.MqMessage;
import com.gengzp.mq.server.model.thread.MqThreadWaiter;
import org.springframework.stereotype.Component;

import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import static com.gengzp.mq.server.constants.MqWaitTimeConstants.WAIT_TIME;

/**
 * @ClassName MqBlockWaitingManagerImpl
 * @Description 消息阻塞等待管理器
 * @Author gengzp
 * @Date 2025/9/3 14:45
 */
@Component
public class MqBlockWaitingManagerImpl implements MqBlockWaitingManager {

    /**
     * 消费者阻塞线程队列映射
     * key:topic    value:阻塞线程等待者队列
     */
    private static final ConcurrentHashMap<String, Queue<MqThreadWaiter>> TOPIC_WAITER_MAP = new ConcurrentHashMap<>();

    @Override
    public MqMessage waitForMessage(String topic, long timeout) throws Exception {
        // 1.创建当前线程的阻塞对象
        MqThreadWaiter mqThreadWaiter = new MqThreadWaiter();
        mqThreadWaiter.setThread(Thread.currentThread());
        mqThreadWaiter.setStartWaitTimeStamp(System.currentTimeMillis());

        // 2.将线程阻塞对象加入到阻塞队列中
        Queue<MqThreadWaiter> mqThreadWaiterQueue = TOPIC_WAITER_MAP.getOrDefault(topic, new ConcurrentLinkedQueue<>());
        mqThreadWaiterQueue.add(mqThreadWaiter);
        TOPIC_WAITER_MAP.put(topic, mqThreadWaiterQueue);

        try {
            // 3.阻塞等待
            synchronized (mqThreadWaiter) {  // 锁当前线程的 mqThreadWaiter 实例
                while (true) {
                    if(mqThreadWaiter.getMqMessage() != null){
                        break;
                    }
                    // 剩余等待时间
                    long remainingWaitTime = WAIT_TIME - System.currentTimeMillis() + mqThreadWaiter.getStartWaitTimeStamp();
                    if (remainingWaitTime > 0) {
                        mqThreadWaiter.wait(remainingWaitTime);
                    } else {
                        break;
                    }
                }
            }

            // 4.唤醒后返回消息
            return mqThreadWaiter.getMqMessage();
        } finally {
            // 5.执行完成, 从队列中移除当前线程阻塞对象
            Queue<MqThreadWaiter> mqThreadWaiters = TOPIC_WAITER_MAP.get(topic);
            if (mqThreadWaiters != null && !mqThreadWaiters.isEmpty()) {
                mqThreadWaiters.remove(mqThreadWaiter);
                // 如果移除后队列为空，则从map中移除
                if (mqThreadWaiters.isEmpty()) {
                    TOPIC_WAITER_MAP.remove(topic);
                }
            }
        }
    }

    @Override
    public void notifyMessageWaiter(MqMessage mqMessage) {
        // 1.获取该topic的等待队列
        Queue<MqThreadWaiter> mqThreadWaiters = TOPIC_WAITER_MAP.get(mqMessage.getTopic());
        if (mqThreadWaiters == null || mqThreadWaiters.isEmpty()) {
            return;
        }

        // 2.唤醒队列中的第一个等待者 (先到先得)
        MqThreadWaiter mqThreadWaiter = mqThreadWaiters.poll();
        if (mqThreadWaiter != null) {
            synchronized (mqThreadWaiter) {
                mqThreadWaiter.setMqMessage(mqMessage);
                mqThreadWaiter.notify();
            }
        }
    }

}
